package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

public class PVStatisBolt implements IRichBolt {

	private OutputCollector _collector;
	private String dip;
	private Integer upstreamoctets;
	private Integer upstreampacket;
	private HashMap<String,Integer> pvMap = new HashMap<String,Integer>();
	private HashMap<String,Integer> usoMap = new HashMap<String,Integer>();
	private HashMap<String,Integer> uspMap = new HashMap<String,Integer>();
	
	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		_collector = collector;
	}

	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		try {

			if(isTickTuple(input)) {//满足时间条件
				emitCountingData();  //计算，入库
				_collector.ack(input);
			} else {//不满足时间条件
				countInLocal(input); //数据记录 
				_collector.ack(input);
			}
		} catch (Exception e) {
			Logger.getRootLogger().error(e);
			throw new FailedException("PVStatisBolt Exception",e);
		}	
	}

	private void countInLocal(Tuple tuple) {
		dip = tuple.getStringByField("dip");      
		upstreamoctets = tuple.getIntegerByField("upstreamoctets");
		upstreampacket = tuple.getIntegerByField("upstreampacket");
		int pv = 1;
		if(pvMap.get(dip) != null) {
			pv = pvMap.get(dip) +1;
		}
		if(usoMap.get(dip) != null) {
			upstreamoctets += usoMap.get(dip);
		}
		if(uspMap.get(dip) != null) {
			upstreampacket += uspMap.get(dip);
		}
		pvMap.put(dip, pv);
		usoMap.put(dip, upstreamoctets);
		uspMap.put(dip, upstreampacket);
	}
	
	private void emitCountingData() {
		SimpleDateFormat sdf_m  = new SimpleDateFormat("yyyyMMddHHmm");
		Date date  = new Date();
		String dateStr = sdf_m.format(date);
		for(Entry<String,Integer> entry : pvMap.entrySet()) {
			String dip = entry.getKey();
			Integer pv = entry.getValue();
			Integer uso = usoMap.get(dip);
			Integer usp = uspMap.get(dip);
			_collector.emit(new Values(dip,dateStr,uso,usp,pv));
		}
		pvMap.clear();
		usoMap.clear();
		uspMap.clear();
	}
	
	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("dip","dateStr","upstreamoctets","upstreampacket","pv"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60);
		return conf;
	}
	
	private boolean isTickTuple(Tuple tuple) {
		return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) && tuple.getSourceStreamId().equals(
				Constants.SYSTEM_TICK_STREAM_ID);   
	}

}
